#include "RelOp.h"

void Join::Run(Pipe &inPipeL, Pipe &inPipeR, Pipe &outPipe, CNF &selOp, Record &literal){

    inL = &inPipeL;
    inR = &inPipeR;
    out = &outPipe;
    cnf = &selOp;
    buffer = &literal;
    joinType = 0;

    FILE *file = fopen("joinType", "r");
    if (file) {
        fscanf(file, "%d", &joinType);
    }
    fclose(file);

    pthread_create(&thread, NULL, JoinWorker, (void*)this);
}

void Join::WaitUntilDone(){
    pthread_join(thread, NULL);
}

void Join::Use_n_Pages(int n){
    nOfPages = n;
}

void Join::DoWork(){

    ComparisonEngine e;
    OrderMaker omL;
    OrderMaker omR;

    if (cnf->GetSortOrders(omL, omR)){

        if (joinType == 0){
             DoHashedWork(omL, omR);
            return;
        }

        if (nOfPages == 0)
            nOfPages = 15;

        Pipe outL(100);
        BigQ qL(*inL, outL, omL, nOfPages);

        Pipe outR(100);
        BigQ qR(*inR, outR, omR, nOfPages);

        Record left, right;
        bool l = outL.Remove(&left);
        bool r = outR.Remove(&right);

        //get all of the records info
        int numAttsLeft = 0;
        int numAttsRight = 0;

        if (left.bits != 0x0)
            numAttsLeft = ((((int*)(left.bits))[1])/sizeof(int)) - 1;
        if (right.bits != 0x0)
            numAttsRight = ((((int*)(right.bits))[1])/sizeof(int)) - 1;

        int numAttsToKeep = numAttsLeft + numAttsRight;
        int attsToKeep[numAttsToKeep];

        for(int i = 0; i < numAttsLeft; i++)
                attsToKeep[i] = i;

        for(int i = 0, j = numAttsLeft; i < numAttsRight; i++, j++)
                attsToKeep[j] = i;

        while(l && r){
            int comp = e.Compare(&left, &omL, &right, &omR);
			if(comp < 0)
                l = outL.Remove(&left);

			else if(comp > 0)
				r = outR.Remove(&right);

			else{
                Record *joinRecord = new Record;
                joinRecord->MergeRecords(&left, &right, numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, numAttsLeft);
                out->Insert(joinRecord);

                vector<Record*> recs;
                Record *copyLeft = new Record;
                copyLeft->Copy(&left);
                recs.push_back(copyLeft);
                int index = 1;

                while (l = outL.Remove(&left)){
                    comp = e.Compare(&left, &omL, &right, &omR);
                    if(comp != 0)
                        break;

                    else{
                        Record *jRecord = new Record;
                        jRecord->MergeRecords(&left, &right, numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, numAttsLeft);
                        out->Insert(jRecord);

                        copyLeft = new Record;
                        copyLeft->Copy(&left);
                        recs.push_back(copyLeft);

                        index++;
                    }
                }

                while (r = outR.Remove(&right)){
                    comp = e.Compare(recs[0], &omL, &right, &omR);
                    if(comp != 0)
                        break;

                    else{
                        int k = 0;
                        do{
                            Record *jRecord = new Record;
                            jRecord->MergeRecords(recs[k], &right, numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, numAttsLeft);
                            out->Insert(jRecord);
                            k++;
                        } while (k < index);
                    }
				}

				for (int i = 0; i < recs.size(); i++)
                    delete (recs[i]);
                recs.clear();
			}
        }
        while (l) l = outL.Remove(&left);
        while (r) r = outR.Remove(&right);
    }
    else {
        DoHardWork();
    }

    out->ShutDown();
}

void Join::DoHashedWork(OrderMaker &omL, OrderMaker &omR){

    multimap<int, Record*> m;

//    OrderMaker omR;
//    omR.numAtts = 1;
//    omR.whichAtts[0] = 1;
//    omR.whichTypes[0] = Int;
//
//    OrderMaker omL;
//    omL.numAtts = 1;
//    omL.whichAtts[0] = 0;
//    omL.whichTypes[0] = Int;

    Record left;
    Record right;

    while (inR->Remove(&right)) {
        Record *r = new Record;
        r->Copy(&right);
        int key = r->ExtractValue(&omR);
        m.insert(multimap<int, Record*>::value_type(key, r));
    }

    inL->Remove(&left);

    //get all of the records info
    int numAttsLeft = 0;
    int numAttsRight = 0;

    if (left.bits != 0x0)
        numAttsLeft = ((((int*)(left.bits))[1])/sizeof(int)) - 1;
    if (right.bits != 0x0)
        numAttsRight = ((((int*)(right.bits))[1])/sizeof(int)) - 1;

    int numAttsToKeep = numAttsLeft + numAttsRight;
    int attsToKeep[numAttsToKeep];

    for(int i = 0; i < numAttsLeft; i++)
            attsToKeep[i] = i;

    for(int i = 0, j = numAttsLeft; i < numAttsRight; i++, j++)
            attsToKeep[j] = i;

    do {
        int key = left.ExtractValue(&omL);
        pair<multimap<int,Record*>::iterator, multimap<int,Record*>::iterator> range;
        range = m.equal_range(key);

        for (multimap<int, Record*>::iterator it = range.first; it != range.second; ++it){
            Record *mergedRecord = new Record;
            mergedRecord->MergeRecords(&left, it->second, numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, numAttsLeft);
            out->Insert(mergedRecord);
        }

    } while (inL->Remove(&left));

    out->ShutDown();
    //TODO: should we clear m and delete pointers??
}

void Join::DoHardWork(){
    vector<Record*> leftVector;
    vector<Record*> rightVector;

    Record left;
    Record right;

    ComparisonEngine e;

    while(inR->Remove(&right)){
        Record *r = new Record;
        r->Copy(&right);
        rightVector.push_back(r);
    }

    inL->Remove(&left);

    //get all of the records info
    int numAttsLeft = ((((int*)(left.bits))[1])/sizeof(int)) - 1;
    int numAttsRight = ((((int*)(rightVector[0]->bits))[1])/sizeof(int)) - 1;
    int numAttsToKeep = numAttsLeft + numAttsRight;
    int attsToKeep[numAttsToKeep];

    for(int i = 0; i < numAttsLeft; i++)
            attsToKeep[i] = i;

    for(int i = 0, j = numAttsLeft; i < numAttsRight; i++, j++)
            attsToKeep[j] = i;


    int bufferMaxSize = 1000;
    int bufferSize = 0;
    int count = 0;
    do{
        if (bufferSize < bufferMaxSize){
            bufferSize++;
            Record *r = new Record;
            r->Copy(&left);
            leftVector.push_back(r);
        }
        else{
            for (int j = 0; j < rightVector.size(); j++){
                for (int i = 0; i < leftVector.size(); i++){
                    if (e.Compare(leftVector[i], rightVector[j], cnf)){
                        Record *joinRecord = new Record;
                        joinRecord->MergeRecords(leftVector[i], rightVector[j], numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, numAttsLeft);
                        out->Insert(joinRecord);
                        count++;
                    }
                }
            }
            leftVector.clear();

            Record *r = new Record;
            r->Copy(&left);
            leftVector.push_back(r);
            bufferSize = 1;
        }
    }while (inL->Remove(&left));

    for (int j = 0; j < rightVector.size(); j++){
        for (int i = 0; i < leftVector.size(); i++){
            if (e.Compare(leftVector[i], rightVector[j], cnf)){
                Record *joinRecord = new Record;
                joinRecord->MergeRecords(leftVector[i], rightVector[j], numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, numAttsLeft);
                out->Insert(joinRecord);
                count++;
            }
        }
    }
    cout<<count<<endl;
    leftVector.clear();
    rightVector.clear();
}

void *JoinWorker(void *param){
     Join join = *((Join*) param);
     join.DoWork();
}
